/*
 * Copyright (c) 2012-2015 The original author or authors
 * ------------------------------------------------------
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Apache License v2.0 which accompanies this distribution.
 *
 * The Eclipse Public License is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * The Apache License v2.0 is available at
 * http://www.opensource.org/licenses/apache2.0.php
 *
 * You may elect to redistribute this code under either of these licenses.
 */
package com.parsechina.hpush.broker.internal;


import com.parsechina.hpush.broker.config.MqttConstants;
import com.parsechina.hpush.broker.internal.listener.ClusterMessaging;
import com.parsechina.hpush.broker.internal.message.NotifyAcknowledgedMessage;
import com.parsechina.hpush.broker.internal.subscriptions.Subscription;
import com.parsechina.hpush.broker.internal.subscriptions.SubscriptionsStore;
import com.parsechina.hpush.broker.parser.*;
import com.parsechina.hpush.broker.spi.*;
import com.parsechina.hpush.broker.utils.ByteUtils;
import com.parsechina.hpush.broker.utils.NettyUtils;
import com.parsechina.hpush.broker.utils.StringUtils;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;
import io.netty.handler.timeout.IdleStateHandler;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;


/**
 * mqtt协议
 */
public class MqttProtocolProcessor implements ProtocolProcessor {

    private static final Pattern COMPILE_SPLIT = Pattern.compile("-", Pattern.LITERAL);
    private static final long serialVersionUID = -2149797748750205852L;

    static final class WillMessage {
        private final String topic;
        private final String payload;
        private final boolean retained;
        private final MqttQoS qos;

        WillMessage(String topic, String payload, boolean retained, MqttQoS qos) {
            this.topic = topic;
            this.payload = payload;
            this.retained = retained;
            this.qos = qos;
        }

        public String getTopic() {
            return topic;
        }

        public String getPayload() {
            return payload;
        }

        boolean isRetained() {
            return retained;
        }

        public MqttQoS getQos() {
            return qos;
        }

    }

    public static class ForceDisconnectEvent {
        private final Channel newChannel;
        private final ConnectMessage msg;

        ForceDisconnectEvent(Channel newChannel, ConnectMessage msg) {
            this.newChannel = newChannel;
            this.msg = msg;
        }

        public Channel getNewChannel() {
            return newChannel;
        }

        public ConnectMessage getMsg() {
            return msg;
        }
    }

    public static class CompleteConnect {
        private final ConnectMessage msg;

        CompleteConnect(ConnectMessage msg) {
            this.msg = msg;
        }

        public ConnectMessage getMsg() {
            return msg;
        }
    }

    private static final Logger LOG = LoggerFactory.getLogger(MqttProtocolProcessor.class);

    /**
     * 连接状态
     */
    public enum ConnectState {
        /**
         * 连接
         */
        CONNECTED,
        /**
         * 断开中
         */
        DISCONNECTING,
        /**
         * 已经断开
         */
        DISCONNECTED,
        /**
         * 删除
         */
        DROP_BY_CONNECT
    }


    private ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
    private SubscriptionsStore subscriptionsStore;
    private boolean allowAnonymous;
    private boolean allowAnonymousClient;
    private boolean allowZeroByteClientId;
    private IMessagesStore messagesStore;
    private ISessionsStore sessionsStore;
    private IAuthenticator authenticator;
    private IClientStore clientStore;
    private IMessaging messaging;

    private final ConcurrentMap<String, WillMessage> willMessageConcurrentMap = new ConcurrentHashMap<>();

    private final ConcurrentMap<String, ConnectState> connectionsStatus = new ConcurrentHashMap<>();


    /**
     * @param subscriptions         the subscription store where are stored all the existing
     *                              clients subscriptionsStore.
     * @param storageService        the persistent store to use for save/load of messages
     *                              for QoS1 and QoS2 handling.
     * @param sessionsStore         the clients sessions store, used to persist subscriptionsStore.
     * @param authenticator         the authenticator used in connect messages.
     * @param allowAnonymous        true connection to clients without credentials.
     * @param allowAnonymousClient  true connection to clients
     * @param allowZeroByteClientId true to allow clients connect without a clientid
     * @param instanceId            cluster instanceId
     * @param redissonClient        redisson client
     */
    public MqttProtocolProcessor(SubscriptionsStore subscriptions,
                                 IMessagesStore storageService,
                                 ISessionsStore sessionsStore,
                                 IAuthenticator authenticator,
                                 IClientStore clientStore,
                                 boolean allowAnonymous,
                                 boolean allowAnonymousClient,
                                 boolean allowZeroByteClientId,
                                 String instanceId,
                                 RedissonClient redissonClient) {

        connectionDescriptors = new ConcurrentHashMap<>();
        this.subscriptionsStore = subscriptions;
        this.allowAnonymous = allowAnonymous;
        this.allowAnonymousClient = allowAnonymousClient;
        this.allowZeroByteClientId = allowZeroByteClientId;
        this.authenticator = authenticator;
        this.messagesStore = storageService;
        this.sessionsStore = sessionsStore;
        this.clientStore = clientStore;
        this.messaging = new ClusterMessaging(redissonClient, this, instanceId);

        LOG.trace("subscription tree on initProtocol {}", subscriptions.dumpTree());
    }

    @Override
    public void processConnect(Channel channel, ConnectMessage msg) {
        LOG.debug("CONNECT for client <{}>", msg.getClientId());
        if (msg.getProtocolVersion() != MqttConstants.VERSION_3_1 && msg.getProtocolVersion() != MqttConstants.VERSION_3_1_1) {
            ConnAckMessage.ConnAckMessageBuilder badProto = ConnAckMessage.buildConnAckMessage();
            badProto.setReturnCode(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION);
            LOG.warn("processConnect sent bad proto ConnAck");
            channel.writeAndFlush(badProto.build());
            channel.close();
            return;
        }

        if (msg.getClientId() == null || msg.getClientId().isEmpty()) {
            if (!msg.isCleanSession() || !allowZeroByteClientId) {

                ConnAckMessage.ConnAckMessageBuilder okResp = ConnAckMessage.buildConnAckMessage();
                okResp.setReturnCode(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED);
                channel.writeAndFlush(okResp.build());
                channel.close();
                return;
            }

            // Generating client id.
//            String randomIdentifier = COMPILE_SPLIT.matcher(UUID.randomUUID().toString()).replaceAll("");
//            msg.setClientId(randomIdentifier);
//            LOG.info("Client connected with server generated identifier: {}", randomIdentifier);
            failedCredentials(channel);
            return;
        }

        // 验证client是否已注册
        if (!allowAnonymousClient) {
            failedCredentials(channel);
            return;
        }

        //handle user authentication
        if (msg.isUserFlag()) {
            String pwd = null;
            if (msg.isPasswordFlag()) {
                pwd = msg.getPassword();
            } else if (!allowAnonymous) {
                failedCredentials(channel);
                return;
            }
            if (!authenticator.checkValid(msg.getClientId(), msg.getUsername(), pwd)) {
                failedCredentials(channel);
                channel.close();
                return;
            }
            NettyUtils.userName(channel, msg.getUsername());
        } else if (!allowAnonymous) {
            failedCredentials(channel);
            return;
        }

        //if an old client with the same ID already exists close its session.
        if (connectionDescriptors.containsKey(msg.getClientId()) /*|| (!otherDisconnecting && existingClientSameId)*/) {
            LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientId());
            //clean the subscriptionsStore if the old used a cleanSession = true
            try {
                Channel oldChannel = connectionDescriptors.get(msg.getClientId()).getChannel();
                closeConnection(oldChannel);

                //注释异步关闭连接事件，防止相同clientId建立和关闭连接同时发生
//                oldChannel.pipeline().fireUserEventTriggered(new ForceDisconnectEvent(channel, msg));
                LOG.debug("Existing connection with same client ID <{}>, forced to close", msg.getClientId());
            } catch (Exception npex) {
                //OK this could happen with concurrent DISCONNECT and CONNECT on same clientId
//                channel.pipeline().fireUserEventTriggered(new CompleteConnect(msg));
                LOG.debug("This could happen with concurrent DISCONNECT and CONNECT on same clientId <{}>", msg.getClientId());
            }
        }

        completeConnect(channel, msg);

    }

    @Override
    public void forceCloseConnection(ConnectMessage msg) {

        if (connectionDescriptors.containsKey(msg.getClientId())) {

            LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientId());
            Channel oldChannel = connectionDescriptors.get(msg.getClientId()).getChannel();

            String clientId = NettyUtils.clientId(oldChannel);
            ClientSession oldClientSession = sessionsStore.sessionForClient(clientId);
            oldClientSession.disconnect();
            NettyUtils.sessionStolen(oldChannel, true);
            NettyUtils.channelStatus(oldChannel, ConnectState.DROP_BY_CONNECT);
            oldChannel.close();

            clientStore.offlineState(clientId);
            sessionsStore.removeSessionOfClient(clientId);
        }

    }

    @Override
    public void forceCloseConnection(Channel oldChannel, Channel newChannel, ConnectMessage msg) {
        String clientId = NettyUtils.clientId(oldChannel);
        ClientSession oldClientSession = sessionsStore.sessionForClient(clientId);
        oldClientSession.disconnect();
        NettyUtils.sessionStolen(oldChannel, true);
        NettyUtils.channelStatus(oldChannel, ConnectState.DROP_BY_CONNECT);
        oldChannel.close();

        newChannel.pipeline().fireUserEventTriggered(new CompleteConnect(msg));

        clientStore.offlineState(clientId);
        sessionsStore.removeSessionOfClient(clientId);
    }

    @Override
    public void closeConnection(Channel oldChannel) {
        String clientId = NettyUtils.clientId(oldChannel);
        ClientSession oldClientSession = sessionsStore.sessionForClient(clientId);
        oldClientSession.disconnect();
        NettyUtils.sessionStolen(oldChannel, true);
        NettyUtils.channelStatus(oldChannel, ConnectState.DROP_BY_CONNECT);
        oldChannel.close();

        clientStore.offlineState(clientId);
        sessionsStore.removeSessionOfClient(clientId);
    }

    @Override
    public void completeConnect(Channel channel, ConnectMessage msg) {
        ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientId(), channel, msg.isCleanSession());
        connectionDescriptors.put(msg.getClientId(), connDescr);

        int keepAlive = msg.getKeepAlive();
        LOG.debug("Connect with keepAlive {} s", keepAlive);
        NettyUtils.keepAlive(channel, keepAlive);
        //session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.isCleanSession());
        NettyUtils.cleanSession(channel, msg.isCleanSession());
        //used to track the client in the subscription and publishing phases.
        //session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(msg.clientId());
        NettyUtils.clientId(channel, msg.getClientId());
        LOG.debug("Connect create session <{}>", channel);

        setIdleTime(channel.pipeline(), Math.round(keepAlive * 1.5f));

        //Handle will flag
        if (msg.isWillFlag()) {
            MqttQoS willQos = MqttQoS.valueOf(msg.getWillQos());
            String willPayload = msg.getWillMessage();

            //save the will testament in the clientId store
            WillMessage will = new WillMessage(msg.getWillTopic(), willPayload, msg.isWillRetain(), willQos);
            willMessageConcurrentMap.put(msg.getClientId(), will);
            LOG.info("Session for clientId <{}> with will to topic {}", msg.getClientId(), msg.getWillTopic());
        }

        ConnAckMessage.ConnAckMessageBuilder okResp = ConnAckMessage.buildConnAckMessage();
        okResp.setReturnCode(MqttConnectReturnCode.CONNECTION_ACCEPTED);

        ClientSession clientSession = sessionsStore.sessionForClient(msg.getClientId());
        boolean isSessionAlreadyStored = clientSession != null;
        if (!msg.isCleanSession() && isSessionAlreadyStored) {
            okResp.setSessionPresent(true);
        }
        if (isSessionAlreadyStored) {
            clientSession.cleanSession(msg.isCleanSession());
        }
        channel.writeAndFlush(okResp.build());

        //通知其他客户端断开连接
        messaging.notifyClientConnected(msg);

        if (!isSessionAlreadyStored) {
            LOG.info("Create persistent session for clientId <{}>", msg.getClientId());
            clientSession = sessionsStore.createNewSession(msg.getClientId(), msg.isCleanSession());
        }

        clientStore.onlineState(msg.getClientId());

        connectionsStatus.put(msg.getClientId(), ConnectState.CONNECTED);
        NettyUtils.channelStatus(channel, ConnectState.CONNECTED);
        if (msg.isCleanSession()) {
            clientSession.cleanSession();
        }
        LOG.info("Connected client ID <{}> with clean session {}", msg.getClientId(), msg.isCleanSession());
        if (!msg.isCleanSession()) {
            //force the republish of stored QoS1 and QoS2
            republishStoredInSession(clientSession);
        }
        //keepAlive * 1000 / 2;
        int flushIntervalMs = 500;
        setupAutoFlusher(channel.pipeline(), flushIntervalMs);
        LOG.info("CONNECT processed");
    }

    private void setupAutoFlusher(ChannelPipeline pipeline, int flushIntervalMs) {
        AutoFlushHandler autoFlushHandler = new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS);
        try {
            pipeline.addAfter("idleEventHandler", "autoFlusher", autoFlushHandler);
        } catch (NoSuchElementException nseex) {
            //the idleEventHandler is not present on the pipeline
            pipeline.addFirst("autoFlusher", autoFlushHandler);
        }
    }

    private void setIdleTime(ChannelPipeline pipeline, int idleTime) {
        if (pipeline.names().contains("idleStateHandler")) {
            pipeline.remove("idleStateHandler");
        }
        pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime));
    }

    private void failedCredentials(Channel session) {

        ConnAckMessage.ConnAckMessageBuilder okResp = ConnAckMessage.buildConnAckMessage();
        okResp.setReturnCode(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
        session.writeAndFlush(okResp.build());
        session.close();
        LOG.info("Client {} failed to connect with bad username or password.", session);
    }

    /**
     * Republish QoS1 and QoS2 messages stored into the session for the clientId.
     */
    private void republishStoredInSession(ClientSession clientSession) {
        LOG.trace("republishStoredInSession for client <{}>", clientSession);
        List<StoredMessage> publishedEvents = clientSession.storedMessages();
        if (publishedEvents.isEmpty()) {
            LOG.info("No stored messages for client <{}>", clientSession.clientId);
            return;
        }

        LOG.info("republishing stored messages to client <{}>", clientSession.clientId);
        for (StoredMessage pubEvt : publishedEvents) {
            //put in flight zone
            LOG.trace("Adding to inflight <{}>", pubEvt.getMessageID());
            clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID());
            directSend(clientSession, pubEvt.getTopic(), pubEvt.getQos(),
                    pubEvt.getMessage(), false, pubEvt.getMessageID());
            clientSession.removeEnqueued(pubEvt.getMessageID(), pubEvt.getGuid());
            clientSession.removeMessageGuid(pubEvt.getMessageID(), pubEvt.getGuid());
        }
    }

    private void verifyToActivate(ClientSession targetSession) {
        if (targetSession == null) {
            return;
        }
        String clientId = targetSession.clientId;
        if (connectionDescriptors.containsKey(clientId)) {
            connectionsStatus.put(clientId, ConnectState.CONNECTED);
        }
    }

    private static StoredMessage asStoredMessage(PublishMessage msg) {
        StoredMessage stored = new StoredMessage(ByteUtils.convert(msg.getPayload()), msg.getQos(), msg.getTopicName());
        stored.setRetained(msg.isRetained());
        stored.setMessageID(msg.getMessageId());
        return stored;
    }

    private static StoredMessage asStoredMessage(WillMessage will) {
        StoredMessage pub = new StoredMessage(ByteUtils.convert(will.getPayload()), will.getQos(), will.getTopic());
        pub.setRetained(will.isRetained());
        return pub;
    }

    @Override
    public void processPublish(Channel channel, PublishMessage msg) {
        LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg);
        String clientId = NettyUtils.clientId(channel);
        final String topic = msg.getTopicName();
        //check if the topic can be wrote
        String username = NettyUtils.userName(channel);

        final MqttQoS qos = msg.getQos();
        final Integer messageId = msg.getMessageId();
        LOG.info("PUBLISH on server from clientId <{}> on topic <{}> with QoS {}", clientId, topic, qos);

        StoredMessage toStoreMsg = asStoredMessage(msg);
        toStoreMsg.setClientId(clientId);
        MessageGUID guid = null;

        if (qos == MqttQoS.AT_MOST_ONCE) { //QoS0
            route2Subscribers(toStoreMsg);
        } else if (qos == MqttQoS.AT_LEAST_ONCE) { //QoS1
            route2Subscribers(toStoreMsg);
            if (msg.isLocal()) {
                sendPubAck(clientId, messageId);
            }
            LOG.info("server replying with PubAck to MSG ID {}", messageId);
        } else if (qos == MqttQoS.EXACTLY_ONCE) { //QoS2
            //store the message
            guid = messagesStore.storePublishForFuture(toStoreMsg);
            if (msg.isLocal()) {
                sendPubRec(clientId, messageId);
            }
            //Next the client will send us a pub rel
            //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
        }

        if (msg.isRetained()) {

            if (qos == MqttQoS.AT_MOST_ONCE || msg.isEmptyPayload()) {

                //QoS == 0 && retain => clean old retained
                //如果服务端收到一条保留（RETAIN）标志为1的QoS 0消息，
                // 它必须丢弃之前为那个主题保留的任何消息。
                //保留标志为1且有效载荷为零字节的PUBLISH报文,
                // 同一个主题下任何现存的保留消息必须被移除
                messagesStore.cleanRetained(topic);

            } else {

                if (guid == null) {
                    //before wasn't stored
                    guid = messagesStore.storePublishForFuture(toStoreMsg);
                }
                messagesStore.storeRetained(topic, guid);
            }
        }

        messaging.notifyTopicPublished(msg, clientId, username);
    }

    /**
     * Intended usage is only for embedded versions of the broker, where the hosting application want to use the
     * broker to send a publish message.
     * Inspired by {@link #processPublish} but with some changes to avoid security check, and the handshake phases
     * for Qos1 and Qos2.
     * It also doesn't notifyTopicPublished because using internally the owner should already know where
     * it's publishing.
     */
    @Override
    public void internalPublish(PublishMessage msg) {


        final MqttQoS qos = msg.getQos();
        final String topic = msg.getTopicName();
        LOG.info("embedded PUBLISH on topic <{}> with QoS {}", topic, qos);

        StoredMessage toStoreMsg = asStoredMessage(msg);

        String clientId = msg.getClientId();
        if (StringUtils.isBlank(clientId)) {
            toStoreMsg.setClientId("BROKER_SELF");
        } else {
            toStoreMsg.setClientId(msg.getClientId());
        }

        toStoreMsg.setMessageID(1);
        MessageGUID guid = null;
        //QoS2
        if (qos == MqttQoS.EXACTLY_ONCE) {
            guid = messagesStore.storePublishForFuture(toStoreMsg);
        }
        route2Subscribers(toStoreMsg);

        if (!msg.isRetained()) {
            return;
        }
        if (qos == MqttQoS.AT_MOST_ONCE || toStoreMsg.isEmptyPayload()) {
            //QoS == 0 && retain => clean old retained
            //如果服务端收到一条保留（RETAIN）标志为1的QoS 0消息，
            // 它必须丢弃之前为那个主题保留的任何消息。
            //保留标志为1且有效载荷为零字节的PUBLISH报文,
            // 同一个主题下任何现存的保留消息必须被移除
            messagesStore.cleanRetained(topic);
            return;
        }
        if (guid == null) {
            guid = messagesStore.storePublishForFuture(toStoreMsg);
        }
        messagesStore.storeRetained(topic, guid);
    }

    @Override
    public boolean isExistClientOfServer(String clientId) {
        ClientSession targetSession = sessionsStore.sessionForClient(clientId);
        return targetSession != null;
    }

    /**
     * Specialized version to publish will testament message.
     */
    private void forwardPublishWill(WillMessage will, String clientID) {
        //it has just to publish the message downstream to the subscribers
        //NB it's a will publish, it needs a PacketIdentifier for this conn, default to 1
        Integer messageId = null;
        if (will.getQos() != MqttQoS.AT_MOST_ONCE) {
            messageId = sessionsStore.nextPacketId(clientID);
        }

        StoredMessage tobeStored = asStoredMessage(will);
        tobeStored.setClientId(clientID);
        tobeStored.setMessageID(messageId);
        route2Subscribers(tobeStored);
    }


    /**
     * 分发消息 (发送者)
     *
     * @param pubMsg 存储的消息
     */
    private void route2Subscribers(final StoredMessage pubMsg) {


        final String topic = pubMsg.getTopic();
        final MqttQoS publishingQos = pubMsg.getQos();
        final ByteBuf origMessage = pubMsg.getMessage();
        LOG.debug("route2Subscribers republishing to existing subscribers that matches the topic {}", topic);
        if (LOG.isDebugEnabled()) {
            LOG.debug("content <{}>", ByteUtils.payload2Str(origMessage));
            LOG.debug("subscription tree {}", subscriptionsStore.dumpTree());
        }

        List<Subscription> topicMatchingSubscriptions = subscriptionsStore.matches(topic);
        LOG.trace("Found {} matching subscriptionsStore to <{}>", topicMatchingSubscriptions.size(), topic);

        //if QoS 1 or 2 store the message
        MessageGUID guid = null;
        StoredMessage storedMessage = null;
        for (final Subscription sub : topicMatchingSubscriptions) {

            storedMessage = pubMsg.cloneMessage();
            MqttQoS qos = publishingQos;
            if (qos.value() > sub.getRequestedQos().value()) {
                qos = sub.getRequestedQos();
            }

            ClientSession targetSession = sessionsStore.sessionForClient(sub.getClientId());
            verifyToActivate(targetSession);

            boolean targetIsActive = connectionsStatus.get(sub.getClientId()) == ConnectState.CONNECTED;

            LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}",
                    sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive);

            ByteBuf message = origMessage.copy();

            if (qos == MqttQoS.AT_MOST_ONCE) {
                //QoS 0
                if (targetIsActive) {
                    directSend(targetSession, topic, MqttQoS.AT_MOST_ONCE, message, false, null);
                }
            } else {

                //QoS 1 or 2
                //if the target subscription is not clean session and is not connected => store it
                if (!targetSession.isCleanSession() && !targetIsActive) {

                    storedMessage.setClientId(sub.getClientId());
                    guid = messagesStore.storePublishForFuture(storedMessage);
                    //store the message in targetSession queue to deliver
                    targetSession.enqueueToDeliver(guid);

                } else {
                    //publish
                    if (targetIsActive) {

                        storedMessage.setClientId(sub.getClientId());
                        guid = messagesStore.storePublishForFuture(storedMessage);

                        int messageId = targetSession.nextPacketId();
                        targetSession.inFlightAckWaiting(guid, messageId);
                        directSend(targetSession, topic, qos, message, false, messageId);
                    }
                }
            }
        }

    }

    private void directSend(ClientSession clientsession, String topic, MqttQoS qos,
                            ByteBuf message, boolean retained, Integer messageId) {
        String clientId = clientsession.clientId;
        LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageId {}",
                clientId, topic, qos, retained, messageId);

        PublishMessage pubMessage = new PublishMessage();

        pubMessage.setRetained(retained);
        pubMessage.setTopicName(topic);
        pubMessage.setQos(qos);
        pubMessage.setPayload(message);

        LOG.info("send publish message to <{}> on topic <{}>", clientId, topic);
        if (LOG.isDebugEnabled()) {
            LOG.debug("content <{}>", ByteUtils.payload2Str(message));
        }
        //set the PacketIdentifier only for QoS > 0
        if (pubMessage.getQos() == MqttQoS.AT_MOST_ONCE) {
            if (messageId != null) {
                throw new RuntimeException("Internal bad error, trying to forwardPublish a QoS 0 message " +
                        "with PacketIdentifier: " + messageId);
            }
        } else {
            pubMessage.setMessageId(messageId);
        }

        if (connectionDescriptors == null) {
            throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be " +
                    "initialized, somewhere it's overwritten!!");
        }
        if (connectionDescriptors.get(clientId) == null) {
            //TODO while we were publishing to the target client, that client disconnected,
            // could happen is not an error HANDLE IT
            throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>",
                    clientId, connectionDescriptors));
        }
        Channel channel = connectionDescriptors.get(clientId).getChannel();
        LOG.trace("Session for clientId {}", clientId);
        if (channel.isWritable()) {
            LOG.debug("channel is writable");
            //if channel is writable don't enqueue
            channel.writeAndFlush(pubMessage.build());
        } else {
            //enqueue to the client session
            LOG.debug("enqueue to client session");
            clientsession.enqueue(pubMessage);
        }
    }

    private void sendPubRec(String clientID, int messageId) {
        LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientId {} with messageID {}", clientID, messageId);
        PubRecMessage pubRecMessage = new PubRecMessage();
        pubRecMessage.setMessageId(messageId);
        connectionDescriptors.get(clientID).getChannel().writeAndFlush(pubRecMessage.build());
    }

    private void sendPubAck(String clientId, int messageId) {
        LOG.trace("sendPubAck invoked");
        PubAckMessage pubAckMessage = new PubAckMessage();
        pubAckMessage.setMessageId(messageId);

        try {
            if (connectionDescriptors == null) {
                throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be initialized, somewhere it's overwritten!!");
            }
            LOG.debug("clientIDs are {}", connectionDescriptors);
            if (connectionDescriptors.get(clientId) == null) {
                throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, connectionDescriptors));
            }
            connectionDescriptors.get(clientId).getChannel().writeAndFlush(pubAckMessage.build());
        } catch (Throwable t) {
            LOG.error(null, t);
        }
    }

    @Override
    public void processPubAck(Channel channel, PubAckMessage msg) {
        String clientId = NettyUtils.clientId(channel);
        int messageId = msg.getMessageId();
        LOG.trace("retrieving inflight for messageId <{}>", messageId);

        //从存储中删除消息
        ClientSession targetSession = sessionsStore.sessionForClient(clientId);
        verifyToActivate(targetSession);

        MessageGUID guid = targetSession.inFlightAcknowledged(messageId);
        targetSession.removeMessageGuid(messageId, guid);

    }

    /**
     * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish
     * to all interested subscribers.
     */
    @Override
    public void processPubRel(Channel channel, PubRelMessage msg) {
        String clientId = NettyUtils.clientId(channel);
        int messageId = msg.getMessageId();
        LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientId {} ad messageId {}", clientId, messageId);
        ClientSession targetSession = sessionsStore.sessionForClient(clientId);
        verifyToActivate(targetSession);
        StoredMessage evt = targetSession.storedMessage(messageId);

        //分发消息
        route2Subscribers(evt);

        targetSession.removeMessageGuid(messageId, evt.getGuid());

        if (evt.isRetained()) {
            final String topic = evt.getTopic();
            if (evt.isEmptyPayload()) {
                messagesStore.cleanRetained(topic);
            }
        }

        sendPubComp(clientId, messageId);
    }

    private void sendPubComp(String clientID, int messageId) {
        LOG.debug("PUB <--PUBCOMP-- SRV sendPubComp invoked for clientId {} ad messageId {}", clientID, messageId);
        PubCompMessage pubCompMessage = new PubCompMessage();
        pubCompMessage.setMessageId(messageId);

        connectionDescriptors.get(clientID).getChannel().writeAndFlush(pubCompMessage.build());
    }

    @Override
    public void processPubRec(Channel channel, PubRecMessage msg) {

        String clientId = NettyUtils.clientId(channel);
        ClientSession targetSession = sessionsStore.sessionForClient(clientId);
        verifyToActivate(targetSession);

        //remove from the inflight and move to the QoS2 second phase queue
        int messageId = msg.getMessageId();
        targetSession.moveInFlightToSecondPhaseAckWaiting(messageId);

        //once received a PUBREC reply with a PUBREL(messageID)
        LOG.debug("\t\tSRV <--PUBREC-- SUB processPubRec invoked for clientId {} ad messageId {}", clientId, messageId);
        PubRelMessage pubRelMessage = new PubRelMessage();
        pubRelMessage.setMessageId(messageId);
        pubRelMessage.setQos(MqttQoS.AT_LEAST_ONCE);

        channel.writeAndFlush(pubRelMessage.build());
    }

    @Override
    public void processPubComp(Channel channel, PubCompMessage msg) {
        String clientId = NettyUtils.clientId(channel);
        int messageId = msg.getMessageId();
        LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientId {} ad messageId {}", clientId, messageId);
        //once received the PUBCOMP then remove the message from the temp memory
        ClientSession targetSession = sessionsStore.sessionForClient(clientId);
        verifyToActivate(targetSession);

        StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageId);
        if (inflightMsg != null) {

            targetSession.removeMessageGuid(messageId, inflightMsg.getGuid());
            String username = NettyUtils.userName(channel);
            String topic = inflightMsg.getTopic();
            messaging.notifyMessageAcknowledged(new NotifyAcknowledgedMessage(inflightMsg, topic, username));
        }

    }

    @Override
    public void processDisconnect(Channel channel, DisconnectMessage msg) {


        channel.flush();
//        channel.writeAndFlush(msg.build());
        ConnectState status = NettyUtils.channelStatus(channel);
        String clientId = NettyUtils.clientId(channel);
        if (status != null && status == ConnectState.DROP_BY_CONNECT) {
            LOG.info("DISCONNECT drop by other connect");
            connectionsStatus.remove(clientId);
            sessionsStore.removeSessionOfClient(clientId);
            clientStore.offlineState(clientId);
            return;
        }
        boolean cleanSession = NettyUtils.cleanSession(channel);
        LOG.info("DISCONNECT client <{}> with clean session {}", clientId, cleanSession);
        boolean notNewConnectionBool = status == null || status != ConnectState.CONNECTED;
        if (notNewConnectionBool) {
            //while disconnect someone else connected
            connectionsStatus.remove(clientId);
            sessionsStore.removeSessionOfClient(clientId);
            clientStore.offlineState(clientId);
            return;
        }
        NettyUtils.channelStatus(channel, ConnectState.DISCONNECTING);
        ClientSession clientSession = sessionsStore.sessionForClient(clientId);
        clientSession.disconnect();
        channel.close();

        status = NettyUtils.channelStatus(channel);
        boolean stillDisconnecting = status != null && status == ConnectState.DISCONNECTING;
        if (!stillDisconnecting) {
            //someone else changed the state (another connect with same client ID)
            return;
        }
        connectionDescriptors.remove(clientId);

        //cleanup the will store
        willMessageConcurrentMap.remove(clientId);

        connectionsStatus.remove(clientId);
        sessionsStore.removeSessionOfClient(clientId);
        clientStore.offlineState(clientId);

        LOG.info("DISCONNECT client <{}> finished", clientId);
    }


    @Override
    public void processConnectionLost(Channel channel, String clientId, boolean sessionStolen) {

        ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientId, channel, true);
        connectionDescriptors.remove(clientId, oldConnDescr);

        if (willMessageConcurrentMap.containsKey(clientId)) {
            WillMessage will = willMessageConcurrentMap.get(clientId);
            forwardPublishWill(will, clientId);
            willMessageConcurrentMap.remove(clientId);
        }

        connectionsStatus.remove(clientId);
        sessionsStore.removeSessionOfClient(clientId);
        clientStore.offlineState(clientId);

        //TODO
//        m_interceptor.notifyClientConnectionLost(clientId, username);
    }

    /**
     * Remove the clientId from topic subscription, if not previously subscribed,
     * doesn't reply any error
     */
    @Override
    public void processUnsubscribe(Channel channel, UnsubscribeMessage msg) {


        List<String> topics = msg.topicFilters();
        int messageId = msg.getMessageId();
        String clientId = NettyUtils.clientId(channel);

        LOG.debug("UNSUBSCRIBE subscription on topics {} for clientId <{}>", topics, clientId);

        ClientSession clientSession = sessionsStore.sessionForClient(clientId);
        verifyToActivate(clientSession);
        for (String topic : topics) {
            boolean validTopic = SubscriptionsStore.validate(topic);
            if (!validTopic) {
                //close the connection, not valid topicFilter is a protocol violation
                channel.close();
                LOG.warn("UNSUBSCRIBE found an invalid topic filter <{}> for clientId <{}>", topic, clientId);
                return;
            }

            subscriptionsStore.removeSubscription(topic, clientId);
            clientSession.unsubscribeFrom(topic);
            String username = NettyUtils.userName(channel);


            //TODO
//            m_interceptor.notifyTopicUnsubscribed(topic, clientId, username);
        }

        //ack the client
        UnsubAckMessage ackMessage = new UnsubAckMessage();
        ackMessage.setMessageId(messageId);

        LOG.info("replying with UnsubAck to MSG ID {}", messageId);
        channel.writeAndFlush(ackMessage.build());
    }

    @Override
    public void processSubscribe(Channel channel, SubscribeMessage msg) {
        String clientId = NettyUtils.clientId(channel);
        LOG.debug("SUBSCRIBE client on server {} packetID {}", clientId, msg.getMessageId());

        ClientSession clientSession = sessionsStore.sessionForClient(clientId);
        verifyToActivate(clientSession);
        //ack the client
        SubAckMessage ackMessage = new SubAckMessage();
        ackMessage.setMessageId(msg.getMessageId());

        String username = NettyUtils.userName(channel);
        List<Subscription> newSubscriptions = new ArrayList<>();
        for (MqttTopicSubscription req : msg.subscriptions()) {
            MqttQoS qos = req.qualityOfService();
            Subscription newSubscription = new Subscription(clientId, req.topicName(), qos);
            boolean valid = clientSession.subscribe(newSubscription);
            ackMessage.addType(valid ? qos : MqttQoS.FAILURE);
            if (valid) {
                newSubscriptions.add(newSubscription);
            }
        }

        //save session, persist subscriptionsStore from session
        LOG.debug("SUBACK for packetID {}", msg.getMessageId());
        if (LOG.isTraceEnabled()) {
            LOG.trace("subscription tree {}", subscriptionsStore.dumpTree());
        }

        for (Subscription subscription : newSubscriptions) {
            subscribeSingleTopic(subscription);
        }
        channel.writeAndFlush(ackMessage.build());

        //fire the persisted messages in session
        for (Subscription subscription : newSubscriptions) {
            publishStoredMessagesInSession(subscription, username);
        }
    }

    private void subscribeSingleTopic(final Subscription newSubscription) {
        LOG.debug("Subscribing {}", newSubscription);
        subscriptionsStore.add(newSubscription.asClientTopicCouple());
    }

    private void publishStoredMessagesInSession(final Subscription newSubscription, String username) {
        LOG.debug("Publish persisted messages in session {}", newSubscription);

        //scans retained messages to be published to the new subscription
        //TODO this is ugly, it does a linear scan on potential big dataset
//        Collection<IMessagesStore.StoredMessage> messages = messagesStore.searchMatching(new IMatchingCondition() {
//            @Override
//            public boolean match(String key) {
//                return SubscriptionsStore.matchTopics(key, newSubscription.getTopicFilter());
//            }
//        });

        Collection<StoredMessage> messages = messagesStore.searchMessage(newSubscription);

        LOG.debug("Found {} messages to republish", messages.size());
        ClientSession targetSession = sessionsStore.sessionForClient(newSubscription.getClientId());
        verifyToActivate(targetSession);

        MqttQoS subQos = newSubscription.getRequestedQos();
        for (StoredMessage storedMsg : messages) {

            MqttQoS qos = storedMsg.getQos();
            if (qos.value() > subQos.value()) {
                qos = subQos;
            }

            //fire as retained the message
            LOG.trace("send publish message for topic {}", newSubscription.getTopicFilter());
            Integer packetID = qos == MqttQoS.AT_MOST_ONCE ? null : targetSession.nextPacketId();
            if (packetID != null) {
                LOG.trace("Adding to inflight <{}>", packetID);
                targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID);
                //服务端发送PUBLISH报文给客户端时，如果消息是作为客户端一个新订阅的结果发送，它必须将报文的保留标志设为1
                directSend(targetSession, storedMsg.getTopic(), qos, storedMsg.getPayload(), true, packetID);
            } else {
                //modify by linfeng add 通知删除消息
                messaging.notifyMessageAcknowledged(new NotifyAcknowledgedMessage(storedMsg, storedMsg.getTopic(), username));
                //服务端发送PUBLISH报文给客户端时，如果消息是作为客户端一个新订阅的结果发送，它必须将报文的保留标志设为1
                directSend(targetSession, storedMsg.getTopic(), qos, storedMsg.getPayload(), true, packetID);
            }

        }

        //notice the Observables
        messaging.notifyTopicSubscribed(newSubscription, username);
    }

    @Override
    public void notifyChannelWritable(Channel channel) {
        String clientID = NettyUtils.clientId(channel);
        ClientSession clientSession = sessionsStore.sessionForClient(clientID);
        boolean emptyQueue = false;
        while (channel.isWritable() && !emptyQueue) {
            PushMessage msg = clientSession.dequeue();
            if (msg == null) {
                emptyQueue = true;
            } else {
                channel.write(msg);
            }
        }
        channel.flush();
    }

}
